package disruptor;


import com.lmax.disruptor.*;

/* This example application uses the ring buffer without the DSL.
 * It creates an event object that contains a single long value,
 * set to the ring buffer sequence number.  The producer sets the
 * event value and publishes the event.  The consumer reads the
 * events and checks the value against the sequence number.
 *
 * Problem encountered:  The send/receive results are inconsistent.
 * Sometimes the consumer receives events that do not match the
 * sequence number provided by the onEvent() argument.  Running the
 * program multiple times gives different results.
 */

public class DisruptorExample {
    RingBuffer<LongEvent> ringBuffer;
    LongEventConsumer eventConsumer;
    LongEventProducer eventProducer;


    public static final int RING_BUFFER_SIZE = 16;
    public static final int MAX_SEQUENCE = 10000;


    public DisruptorExample() {
    }

    public static void main(String args[]) {
        DisruptorExample app = new DisruptorExample();
        app.appMain(args);
        //System.exit();
    }


    public int appMain(String args[]) {
        this.makeRingBuffer();
        this.makeConsumer();
        this.makeProducer();

        try {
            this.eventProducer.thread.join();
            this.eventConsumer.thread.join();
            //Thread.sleep(1000);
        } catch (Exception e) {
            // do nothing
            System.out.println("ssssssssssssssssss");
        }
        return 0;
    }


    public void makeConsumer() {
        this.eventConsumer = new LongEventConsumer(this.ringBuffer);
    }


    public void makeProducer() {
        this.eventProducer = new LongEventProducer(this.ringBuffer);
    }


    public void makeRingBuffer() {
        this.ringBuffer = RingBuffer.createSingleProducer(
                new LongEventFactory(),
                RING_BUFFER_SIZE,
                new SleepingWaitStrategy()
        );
    }


    public class LongEvent {
        /* Value is set to ring buffer sequence number.
         */
        public long value = 0;
    }


    public class LongEventFactory
            implements EventFactory<LongEvent> {
        public LongEvent newInstance() {
            return new LongEvent();
        }
    }


    public class LongEventConsumer
            implements EventHandler<LongEvent> {
        BatchEventProcessor<LongEvent> eventProcessor;
        public Thread thread;

        public LongEventConsumer(RingBuffer<LongEvent> ringBuffer) {
            Sequence[] sequences = new Sequence[0];
            SequenceBarrier barrier = ringBuffer.newBarrier(sequences);
            this.eventProcessor = new BatchEventProcessor<LongEvent>(
                    ringBuffer,
                    barrier,
                    this
            );
            ringBuffer.addGatingSequences(eventProcessor.getSequence());
            this.runThread();
        }

        public void runThread() {
            this.thread = new Thread(this.eventProcessor);
            this.thread.setName(
                    this.getClass().getSimpleName()
                            + "-Consumer-"
                            + thread.getId()
            );

            /* Set the thread to be a daemon.  When the rest of the application
             * is ready to terminate, this producer thread is expendable.  We do
             * not want the runtime system to wait for this thread to terminate.
             */

            this.thread.setDaemon(true);
            this.thread.start();
        }


        public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
            /* Note the provided event object "belongs" to the ring buffer.
             * We must copy it for downstream use.
             */

            if (event.value != sequence) {
                System.out.printf("consumer value should be %d but is %d\n", sequence, event.value);
                this.eventProcessor.halt();
            }
            if (event.value >= MAX_SEQUENCE || sequence >= MAX_SEQUENCE) {
                this.eventProcessor.halt();
            }
        }
    }


    public class LongEventProducer {
        private final RingBuffer<LongEvent> ringBuffer;
        public Thread thread;

        public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
            this.runThread();
        }

        public void injectValue() {
            long sequence = this.ringBuffer.next();
            try {
                LongEvent event = ringBuffer.get(sequence);
                event.value = sequence;
            } finally {
                this.ringBuffer.publish(sequence);
            }
        }


        public void runThread() {
            Runnable runnable = new Runnable() {
                public void run() {
                    for (int j = 0; j <= MAX_SEQUENCE; ++j) {
                        if (Thread.currentThread().isInterrupted()) {
                            break;
                        }
                        LongEventProducer.this.injectValue();
                    }
                }
            };
            this.thread = new Thread(runnable);
            this.thread.setName(
                    this.getClass().getSimpleName()
                            + "-Producer-"
                            + thread.getId()
            );

            /* Set the thread to be a daemon.  When the rest of the application
             * is ready to terminate, this producer thread is expendable.  We do
             * not want the runtime system to wait for this thread to terminate.
             */

            this.thread.setDaemon(true);
            this.thread.start();
        }
    }
}